package com.orkasgb.framework.hadoop.mapperreduce.join.mapperjoin;

import org.apache.commons.io.FileUtils;
import org.apache.commons.io.LineIterator;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;

public class JoinMapper extends Mapper<LongWritable, Text, NullWritable, JoinEntity> {

    private final Map<String, String> cacheMap = new HashMap<>();

    @Override
    protected void setup(Mapper<LongWritable, Text, NullWritable, JoinEntity>.Context context) throws IOException {

        URI[] cacheFiles = context.getCacheFiles();
        LineIterator strings = FileUtils.lineIterator(new File(cacheFiles[0].getPath()));
        while (strings.hasNext()) {
            String line = strings.nextLine();
            String[] split = line.split("\t");
            cacheMap.put(split[1], split[2]);
        }
    }

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, NullWritable, JoinEntity>.Context context) throws IOException, InterruptedException {
        if (key.get() == 0) {
            return;
        }

        String[] split = value.toString().split("\t");
        JoinEntity joinEntity = new JoinEntity();
        String id = split[1];
        joinEntity.setNo(Integer.parseInt(split[0]));
        joinEntity.setId(id);
        joinEntity.setNum(Integer.parseInt(split[2]));
        joinEntity.setDate(split[3]);

        joinEntity.setName(cacheMap.get(id));
        joinEntity.setflag("");
        context.write(NullWritable.get(), joinEntity);
    }
}
